// Copyright (c) 2000 Just Objects B.V. <just@justobjects.nl>
// Distributable under LGPL license. See terms of license at gnu.org.

package nl.justobjects.pushlet.core;

import nl.justobjects.pushlet.util.Log;
import nl.justobjects.pushlet.util.PushletException;

import javax.servlet.http.Cookie;
import javax.servlet.http.HttpServletRequest;
import java.io.IOException;
import java.net.URLDecoder;

/**
 * Handles servlet requests from client.
 *
 * @author Just van den Broecke - Just Objects &copy;
 * @version $Id: Controller.java,v 1.9 2007/11/23 14:33:07 justb Exp $
 */
public class Controller implements Protocol, ConfigDefs {

	// private Session session;

	/**
	 * Protected constructor as we create through factory method.
	 */
	protected Controller() {
	}

	/**
	 * Create instance through factory method.
	 *
	 * @param aSession the parent Session
	 * @return a Controller object (or derived)
	 * @throws PushletException exception, usually misconfiguration
	 */
	/*public static Controller create(Session aSession) throws PushletException {
		Controller controller;
		try {
			controller = (Controller) Config.getClass(CONTROLLER_CLASS, "nl.justobjects.pushlet.core.Controller").newInstance();
		} catch (Throwable t) {
			throw new PushletException("Cannot instantiate Controller from config", t);
		}
		controller.session = aSession;
		return controller;
	}*/

	/**
	 * Handle command.
	 */
	public static void doCommand(Command aCommand) {
		Session session = aCommand.getSession();
		try {
			// Update lease time to live
			session.kick();

			// Set remote IP address of client
			session.setAddress(aCommand.httpReq.getRemoteAddr());

			// Get event type
			String eventType = aCommand.reqEvent.getEventType();
            debug("doCommand() event=" + eventType);

			// Determine action based on event type
			if (eventType.equals(Protocol.E_REFRESH)) {
				// Pull/poll mode clients that refresh
				doRefresh(aCommand);
			} else if (eventType.equals(Protocol.E_SUBSCRIBE)) {
				// Subscribe
				doSubscribe(aCommand);
			} else if (eventType.equals(Protocol.E_UNSUBSCRIBE)) {
				// Unsubscribe
				doUnsubscribe(aCommand);
			} else if (eventType.equals(Protocol.E_JOIN)) {
				// Join
				doJoin(aCommand);
            } else if (eventType.equals(Protocol.E_LISTEN)) {
                // Listen to pushed events
                doListen(aCommand);
			} else if (eventType.equals(Protocol.E_JOIN_LISTEN)) {
				// Join and listen (for simple and e.g. REST apps)
				doJoinListen(aCommand);
			} else if (eventType.equals(Protocol.E_LEAVE)) {
				// Leave
				doLeave(aCommand);
			} else if (eventType.equals(Protocol.E_HEARTBEAT)) {
				// Heartbeat mainly to do away with browser "busy" cursor
				doHeartbeat(aCommand);
			} else if (eventType.equals(Protocol.E_PUBLISH)) {
				// Publish event
				doPublish(aCommand);
			} else if (eventType.equals(Protocol.E_RECONNECT)) {
                doReconnect(aCommand);
            }

			// Handle response back to client
			if ( eventType.equals(Protocol.E_REFRESH) || eventType.endsWith(Protocol.E_LISTEN) ||
                    eventType.equals(Protocol.E_RECONNECT)) {
				// Data channel events
				// Loops until refresh or connection closed
				getSubscriber(aCommand).fetchEvents(aCommand);
			} else {
				// Send response for control commands
				sendControlResponse(aCommand);
			}

		} catch (Throwable t) {
			warn("Exception in doCommand(): " + t);
			t.printStackTrace();
		}
	}

	/**
	 * Handle heartbeat event.
	 */
	protected static void doHeartbeat(Command aCommand) {

		// Set heartbeat acknowledgement to client
		aCommand.setResponseEvent(new Event(E_HEARTBEAT_ACK));
	}

    /**
     * 处理reconnect
     */
    protected static void doReconnect(Command aCommand) throws PushletException {
        Event responseEvent = null;
        Session session = aCommand.getSession();
        try {
            // join
            String format = aCommand.reqEvent.getField(P_FORMAT, Config.getProperty(EVENT_FORMAT, FORMAT_JSON));
            session.start();
            session.setFormat(format);
            // listen
            String mode = (String) aCommand.reqEvent.getField(P_MODE);
            if (mode == null || Config.getBoolProperty(LISTEN_FORCE_PULL_ALL)) {
                mode = MODE_PULL;
            } else {
            	if (!MODE_PULL.equals(mode)) {
	                String userAgent = session.getUserAgent();
	                for (int i = 0, size = Session.FORCED_PULL_AGENTS.length; i < size; i++) {
	                    if ((userAgent.indexOf(Session.FORCED_PULL_AGENTS[i]) != -1)) {
	                        info("Forcing pull mode for agent=" + userAgent);
	                        mode = MODE_PULL;
	                        break;
	                    }
	                }
            	}
            }
            getSubscriber(aCommand).setMode(mode);
            getSubscriber(aCommand).start();
            // subscribe
            HttpServletRequest request = (HttpServletRequest) aCommand.httpReq;
            Cookie[] cookies = request.getCookies();
            String cookieValue = null;
            if (cookies != null) {
            	for (int i = 0, size = cookies.length; i < size; i++) {
            		if (Protocol.COOKIE_NAME.equalsIgnoreCase(cookies[i].getName())) {
            			cookieValue = cookies[i].getValue();
            			break;
            		}
            	}
            }
            if (cookieValue != null) {
            	cookieValue = URLDecoder.decode(cookieValue, "UTF-8");
                String[] subjectAndLabels = cookieValue.split(Protocol.SUBJECTS_GAP);
                for (int i = 0, size = subjectAndLabels.length; i < size; i++) {
                    String[] subjectAndLabel = subjectAndLabels[i].split(Protocol.SUBJECT_LABEL_GAP);
                    getSubscriber(aCommand).addSubscription(subjectAndLabel[0], (subjectAndLabel.length == 2 ? subjectAndLabel[1] : null));
                }
            }
            responseEvent = new ReconnectAckEvent();
            responseEvent.setField(P_ID, session.getId());
            responseEvent.setField(P_FORMAT, format);
            responseEvent.setField(P_MODE, mode);
            info("Reconnect");
        } catch (Throwable t) {
            session.stop();
            responseEvent = new NackEvent();
            responseEvent.setField(P_ID, session.getId());
            responseEvent.setField(P_REASON, "unexpected error: " + t);
            warn("doReconnect() error: " + t);
            t.printStackTrace();
        } finally {
            aCommand.setResponseEvent(responseEvent);
        }
    }

	/**
	 * Handle Join request.
	 */
	protected static void doJoin(Command aCommand) throws PushletException {

		Event responseEvent = null;
		Session session = aCommand.getSession();
		try {
			String format = aCommand.reqEvent.getField(P_FORMAT, FORMAT_XML_STRICT);
			session.start();
			// Determine format for encoding Events to client.
			// Default assume a userAgent window on the other end.
			session.setFormat(format);
			
			responseEvent = new Event(E_JOIN_ACK);
			// Set unique subscriber id and encoding format
			responseEvent.setField(P_ID, session.getId());
			responseEvent.setField(P_FORMAT, format);
			info("joined");
		} catch (Throwable t) {
			session.stop();
			responseEvent = new NackEvent();
			responseEvent.setField(P_ID, session.getId());
			responseEvent.setField(P_REASON, "unexpected error: " + t);
			warn("doJoin() error: " + t);
			t.printStackTrace();
		} finally {
			// Always set response event in command
			aCommand.setResponseEvent(responseEvent);
		}

	}

	/**
	 * Handle JoinListen request.
	 */
	protected static void doJoinListen(Command aCommand) throws PushletException {

		// Basically bundles a join and a listen
		// This request is handly for simple apps that
		// need to do a single request to get events immediately
		// For example in RESTful apps.

		// First do regular join
		doJoin(aCommand);
		if (!aCommand.getResponseEvent().getEventType().equals(E_NACK)) {
			// If successful do the listen
			doListen(aCommand);
			if (!aCommand.getResponseEvent().getEventType().equals(E_NACK)) {
				// If still ok do the listen ack
				aCommand.getResponseEvent().setField(P_EVENT, E_JOIN_LISTEN_ACK);
			}
		}
	}

	/**
	 * Handle Leave request.
	 */
	protected static void doLeave(Command aCommand) throws IOException {

		Event responseEvent = null;
		Session session = aCommand.getSession();
		try {
			// Also removes all subscriptions
			session.stop();

            // Prepare acknowledgement
			responseEvent = new LeaveAckEvent();

			// Set unique subscriber id
			responseEvent.setField(P_ID, session.getId());
			info("left");
		} catch (Throwable t) {
			responseEvent = new NackEvent();
			responseEvent.setField(P_ID, session.getId());
			responseEvent.setField(P_REASON, "unexpected error: " + t);
			warn("doLeave() error: " + t);
			t.printStackTrace();
		} finally {
			// Always set response event in command
			aCommand.setResponseEvent(responseEvent);
		}

	}

	/**
	 * Handle Listen request.
	 */
	protected static void doListen(Command aCommand) throws PushletException {
		Session session = aCommand.getSession();
        if (getSubscriber(aCommand).isActive()) {
            warn("Already under listening. P_id=" + session.getId() + ", Subject will not be subcribed.");
            Event listenAckEvent = new Event(E_LISTEN_ACK);
            listenAckEvent.setField(P_ID, session.getId());
            listenAckEvent.setField(P_MODE, getSubscriber(aCommand).getMode());
            listenAckEvent.setField(P_FORMAT, session.getFormat());
            aCommand.setResponseEvent(listenAckEvent);
            return;
        }

		String mode = MODE_STREAM;
		// Should we always force "pull" mode ? PushetClient只支持stream方式
		if (Config.getBoolProperty(LISTEN_FORCE_PULL_ALL)) {
			mode = MODE_PULL;
		} else {
			// Determine optimal mode determined by parameter and/or user agent
			// Mode param determines how events are transfered to the client

			// In "stream" mode, a stream of events is sent, i.e. the document
			// is neverending. In "pull" or "poll" mode a complete document is returned
			// ending with a request to refresh.
			mode = aCommand.reqEvent.getField(P_MODE, MODE_STREAM);
			String userAgent = aCommand.httpReq.getHeader("User-Agent");
			if (userAgent != null) {
				userAgent = userAgent.toLowerCase();
				if (!MODE_PULL.equals(mode)) {
					for (int i = 0; i < Session.FORCED_PULL_AGENTS.length; i++) {
						if ((userAgent.indexOf(Session.FORCED_PULL_AGENTS[i]) != -1)) {
							info("Forcing pull mode for agent=" + userAgent);
							mode = MODE_PULL;
							break;
						}
					}
				}
			} else {
				userAgent = "unknown";
			}
		}

		getSubscriber(aCommand).setMode(mode);

		// Prepare acknowledgement
		Event listenAckEvent = new Event(E_LISTEN_ACK);

		// Add subscription(s) if subject(s) specified
		String subject = aCommand.reqEvent.getSubject();
		if (subject != null) {
			// Optional label for subscription
			String label = (String) aCommand.reqEvent.getField(Protocol.P_LABEL);

			// Add a subscription
			/*Subscription subscription = */
            getSubscriber(aCommand).addSubscription(subject, label);

			// Add subscription subject, id and optional label to listen-ack event
            listenAckEvent.setField(P_SUBJECT, subject);
			// listenAckEvent.setField(P_SUBSCRIPTION_ID, subscription.getId());
			if (label != null) {
				listenAckEvent.setField(P_LABEL, label);
			}
		}

		// Set unique subscriber id, push mode and encoding format
		listenAckEvent.setField(P_ID, session.getId());
		listenAckEvent.setField(P_MODE, mode);
		listenAckEvent.setField(P_FORMAT, session.getFormat());

		// Activate the subscriber
		getSubscriber(aCommand).start();

		// Enqueue listen ack event on data channel
		aCommand.setResponseEvent(listenAckEvent);

		info("Listening mode=" + mode + " userAgent=" + session.getUserAgent());
	}

	/**
	 * Handle Publish request.
	 */
	// TODO 通过脚本发送消息（走JMS）
	protected static void doPublish(Command aCommand) {
		Event responseEvent = null;
		Session session = aCommand.getSession();
		try {
			aCommand.reqEvent.setField(P_FROM, session.getId());
			aCommand.reqEvent.setField(P_EVENT, E_DATA);
			String subject = aCommand.reqEvent.getSubject();
			if (subject == null) {
				// Return error response
				// responseEvent = new NackEvent();
				// responseEvent.setField(P_ID, session.getId());
				// responseEvent.setField(P_REASON, "no subject provided");
				Dispatcher.getInstance().broadcast(aCommand.reqEvent);
			} else {
				/*aCommand.reqEvent.setField(P_FROM, session.getId());
				aCommand.reqEvent.setField(P_EVENT, E_DATA);*/

				// Event may be targeted to specific user (p_to field)
				String to = (String) aCommand.reqEvent.getField(P_TO);
				if (to != null) {
					Dispatcher.getInstance().unicast(aCommand.reqEvent, to);
				} else {
					// No to: multicast
					// debug("doPublish() event=" + aCommand.reqEvent);
					Dispatcher.getInstance().multicast(aCommand.reqEvent);
				}

				// Acknowledge
				// responseEvent = new Event(E_PUBLISH_ACK);
			}
			responseEvent = new Event(E_PUBLISH_ACK);
		} catch (Throwable t) {
			responseEvent = new NackEvent();
			responseEvent.setField(P_ID, session.getId());
			responseEvent.setField(P_REASON, "unexpected error: " + t);
			warn("doPublish() error: " + t);
			t.printStackTrace();
		} finally {
			// Always set response event in command
			aCommand.setResponseEvent(responseEvent);
		}
	}

	/**
	 * Handle refresh event.
	 */
	protected static void doRefresh(Command aCommand) {
		// Set ack
		aCommand.setResponseEvent(new RefreshAckEvent());
	}

	/**
	 * Handle Subscribe request.
	 */
	protected static  void doSubscribe(Command aCommand) throws IOException {
		Session session = aCommand.getSession();
		Event responseEvent = null;
		try {
			String subject = aCommand.reqEvent.getSubject();
			Subscription subscription = null;
			if (subject == null) {
				// Return error response
				responseEvent = new NackEvent();
				responseEvent.setField(P_ID, session.getId());
				responseEvent.setField(P_REASON, "no subject provided");
			} else {

				String label = (String) aCommand.reqEvent.getField(Protocol.P_LABEL);
                subscription = getOldSubscription(subject, label, aCommand);
                if (subscription == null)
				    subscription = getSubscriber(aCommand).addSubscription(subject, label);
                else {
                	// 推送本主题的最新一条消息
                	Event lastestEvent = EventCache.getInstance().getEvent(subscription.getCompositeSubject());
                	if (lastestEvent != null) {
                		session.getSubscriber().onEvent(lastestEvent);
                	}
                	warn("Already Subscribed. P_id=" + session.getId() + ", Old Subscription will be used.");
                }

				// Acknowledge
				responseEvent = new SubscribeAckEvent();
				responseEvent.setField(P_ID, session.getId());
				responseEvent.setField(P_SUBJECT, subject);
				// responseEvent.setField(P_SUBSCRIPTION_ID, subscription.getId());
				if (label != null) {
					responseEvent.setField(P_LABEL, label);
				}
				info("subscribed to " + subject/* + " sid=" + subscription.getId()*/ + " label=" + label);
			}

		} catch (Throwable t) {
			responseEvent = new NackEvent();
			responseEvent.setField(P_ID, session.getId());
			responseEvent.setField(P_REASON, "unexpected error: " + t);
			warn("doSubscribe() error: " + t);
			t.printStackTrace();
		} finally {
			// Always set response event in command
			aCommand.setResponseEvent(responseEvent);
		}
	}

    protected static  Subscription getOldSubscription(String aSubject, String aLabel, Command aCommand) {
        Subscriber subscriber = aCommand.getSession().getSubscriber();
        return subscriber.getSubscription(aSubject, aLabel);
    }

	/**
	 * Handle Unsubscribe request.
	 */
	protected static void doUnsubscribe(Command aCommand) throws IOException {

		Session session = aCommand.getSession();
		Event responseEvent = null;
		try {
			String subject = aCommand.reqEvent.getSubject();
			if (subject == null) {
				// Unsuscbribe all
				getSubscriber(aCommand).removeSubscriptions();
				responseEvent = new UnsubscribeAckEvent();
				responseEvent.setField(P_ID, session.getId());
				info("unsubscribed all");
			} else {
				String label = (String) aCommand.reqEvent.getField(Protocol.P_LABEL);
				// Subscription id provided: remove Subscription
                Subscription subscription = getOldSubscription(subject, label, aCommand);
                
                // OK return ack
				responseEvent = new UnsubscribeAckEvent();
				responseEvent.setField(P_ID, session.getId());
				if (subscription != null) {
					getSubscriber(aCommand).removeSubscription(subscription.getCompositeSubject());
				}
				responseEvent.setField(P_SUBJECT, subject);
				if (label != null) {
					responseEvent.setField(P_LABEL, label);
				}
				info("unsubscribed subject=" + subject + " label=" + label);
			}
		} catch (Throwable t) {
			responseEvent = new NackEvent();
			responseEvent.setField(P_ID, session.getId());
			responseEvent.setField(P_REASON, "unexpected error: " + t);
			warn("doUnsubscribe() error: " + t);
			t.printStackTrace();
		} finally {
			// Always set response event in command
			aCommand.setResponseEvent(responseEvent);
		}
	}

	public static Subscriber getSubscriber(Command aCommand) {
		return aCommand.getSession().getSubscriber();
	}

	/**
	 * Send response on the control channel.
	 */
	protected static void sendControlResponse(Command aCommand) {
		try {

			// Try to prevent caching in any form.
			aCommand.sendResponseHeaders();

			// Let clientAdapter determine how to send event
			aCommand.getClientAdapter().start();

			// Push to client through client adapter
			aCommand.getClientAdapter().push(aCommand.getResponseEvent());

			// One shot response
			aCommand.getClientAdapter().stop();
        } catch (IOException e) {
            e.printStackTrace();
		} catch (Throwable t) {
			t.printStackTrace();
			aCommand.getSession().stop();
		}
	}


	/**
	 * Info.
	 */
	protected static void info(String s) {
		Log.info("[Controller] " + s);
	}

	/**
	 * Exceptional print util.
	 */
	protected static void warn(String s) {
		Log.warn("[Controller] " + s);
	}

	/**
	 * Exceptional print util.
	 */
	protected static void debug(String s) {
		Log.debug("[Controller] " + s);
	}


}

/*
 * $Log: Controller.java,v $
 * Revision 1.9  2007/11/23 14:33:07  justb
 * core classes now configurable through factory
 *
 * Revision 1.8  2005/02/28 15:58:05  justb
 * added SimpleListener example
 *
 * Revision 1.7  2005/02/28 13:05:59  justb
 * introduced join-listen protocol service
 *
 * Revision 1.6  2005/02/28 12:45:59  justb
 * introduced Command class
 *
 * Revision 1.5  2005/02/28 09:14:55  justb
 * sessmgr/dispatcher factory/singleton support
 *
 * Revision 1.4  2005/02/25 15:13:00  justb
 * session id generation more robust
 *
 * Revision 1.3  2005/02/21 16:59:06  justb
 * SessionManager and session lease introduced
 *
 * Revision 1.2  2005/02/21 12:32:28  justb
 * fixed publish event in Controller
 *
 * Revision 1.1  2005/02/21 11:50:46  justb
 * ohase1 of refactoring Subscriber into Session/Controller/Subscriber
 *

 *
 */
